package cn.tedu.knows.kafka.demo;

import cn.tedu.knows.kafka.vo.Message;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class Consumer {

    // 我们添加的kafka依赖中有一个kafka监听器类
    // kafka监听器负责在有消息进入kafka时通知对应的话题消费者来接收
    // 下面的注解就是指定kafka监听器监听的话题名称
    @KafkaListener(topics = "myTopic")
    public void receive(
                ConsumerRecord<String,String> record){
        // 方法上面的注解效果是当myTopic话题有新消息时
        // 会自动调用下面的方法,方法的参数就是消息对象
        // ConsumerRecord<[话题名称类型],[消息类型]>

        // 获得消息具体内容
        String json=record.value();
        // 将json格式字符串转换成java对象
        Gson gson=new Gson();
        Message message=gson.fromJson(json,Message.class);
        // 日志输出记录
        log.debug("接收到消息{}",message);
    }

}









